/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package remoting.exchange.support;


import static common.constants.CommonConstants.DEFAULT_REMOTING_SERIALIZATION;
import static common.constants.CommonConstants.SERIALIZATION_KEY;
import static serialization.Constants.COMPACTED_JAVA_SERIALIZATION_ID;
import static serialization.Constants.JAVA_SERIALIZATION_ID;
import static serialization.Constants.NATIVE_JAVA_SERIALIZATION_ID;

import common.model.MagicUrl;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import serialization.Impl.Hessian2Serialization;
import serialization.ObjectInput;
import serialization.Serializer;

public class CodecSupport {

    private static final Logger logger = LoggerFactory.getLogger(CodecSupport.class);
    private static Map<Byte, Serializer> ID_Serializer_MAP = new HashMap<Byte, Serializer>();
    private static Map<Byte, String> ID_SerializerNAME_MAP = new HashMap<Byte, String>();
    private static Serializer Serializer = new Hessian2Serialization();
    static {
        byte idByte = Serializer.getContentTypeId();
        if (ID_Serializer_MAP.containsKey(idByte)) {
            logger.error("Serializer extension " + Serializer.getClass().getName()
                + " has duplicate id to Serializer extension "
                + ID_Serializer_MAP.get(idByte).getClass().getName()
                + ", ignore this Serializer extension");
        }
        ID_Serializer_MAP.put(idByte, Serializer);
        ID_SerializerNAME_MAP.put(idByte, "hession2");
    }

    private CodecSupport() {
    }

    public static Serializer getSerializerById(Byte id) {
        return ID_Serializer_MAP.get(id);
    }

    public static Serializer getSerializer(MagicUrl url) {
        return Serializer;
    }

    public static Serializer getSerializer(MagicUrl url, Byte id) throws IOException {
        Serializer Serializer = getSerializerById(id);
        return Serializer;
    }

    public static ObjectInput deserialize(MagicUrl url, InputStream is, byte proto) throws IOException {
        Serializer s = getSerializer(url, proto);
        return s.deserialize(url, is);
    }


    public static Serializer getSerialization(MagicUrl url, Byte id) throws IOException {
        Serializer serialization = getSerializerById(id);
        String serializationName = url.getParameter(SERIALIZATION_KEY, DEFAULT_REMOTING_SERIALIZATION);
        // Check if "serialization id" passed from network matches the id on this side(only take effect for JDK serialization), for security purpose.
        if (serialization == null
            || ((id == JAVA_SERIALIZATION_ID || id == NATIVE_JAVA_SERIALIZATION_ID || id == COMPACTED_JAVA_SERIALIZATION_ID)
            && !(serializationName.equals(ID_SerializerNAME_MAP.get(id))))) {
            throw new IOException("Unexpected serialization id:" + id + " received from network, please check if the peer send the right id.");
        }
        return serialization;
    }
}
